package com.arch.flink.archetype.network;

import com.arch.flink.archetype.network.handler.InboundEnvelopeDecoder;
import com.arch.flink.archetype.network.handler.InboundEnvelopeDispatcher;
import org.apache.flink.shaded.netty4.io.netty.bootstrap.ServerBootstrap;
import org.apache.flink.shaded.netty4.io.netty.buffer.PooledByteBufAllocator;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInitializer;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelOption;
import org.apache.flink.shaded.netty4.io.netty.channel.FixedRecvByteBufAllocator;
import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoopGroup;
import org.apache.flink.shaded.netty4.io.netty.channel.socket.SocketChannel;
import org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioServerSocketChannel;

import java.io.IOException;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.UnknownHostException;

/**
 * @author pizhihui
 * @date 2024-05-20 18:02
 */
public class NettyServer {


    private ServerBootstrap in;


    private final int numInThreads = 10;
    private final int numOutThreads = 10;

    private final int bufferSize = 2048;
    private final int pageSize = 1024;



    private final InetAddress bindAddress = Inet4Address.getLocalHost();
    private final int bindPort = 4096;


    public NettyServer() throws UnknownHostException {
        // 参数

    }


    public void start() throws IOException {

        int numHeapArenas = 0;
        int numDirectArenas = numInThreads + numOutThreads;
        int pageSize = bufferSize << 1;
        int chunkSize = 16 * 1 << 20; // 16 MB
        int maxOrder = (int) (Math.log(chunkSize/pageSize) / Math.log(2));


        //  pageSize: 2048 (expected: 4096)
        PooledByteBufAllocator pooledByteBufAllocator =
                new PooledByteBufAllocator(true, numHeapArenas, numDirectArenas, pageSize, maxOrder);

        // --------------------------------------------------------------------
        // server bootstrap (incoming connections)
        // --------------------------------------------------------------------
        in = new ServerBootstrap();
        in.group(new NioEventLoopGroup(numInThreads))
                .channel(NioServerSocketChannel.class)
                .localAddress(bindAddress, bindPort)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline()
                                .addLast(new InboundEnvelopeDecoder())
                                .addLast(new InboundEnvelopeDispatcher());
                    }
                })
                .option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(pageSize))
                .option(ChannelOption.ALLOCATOR, pooledByteBufAllocator);

        try {
            in.bind().sync();
        } catch (InterruptedException e) {
            throw new IOException(e);
        }

        System.out.println("服务端启动完成....");

    }

    public static void main(String[] args) throws IOException {
        NettyServer server = new NettyServer();
        server.start();
    }

}
